package com.hadooop.topN;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Comparator;
import java.util.PriorityQueue;

/*
这里优先队列里换成存储Integer就没问题了
 */
public class TopNReducer2 extends Reducer<IntWritable, NullWritable, NullWritable, IntWritable> {

    private int N = 5;

    private PriorityQueue<Integer> topNQueue = new PriorityQueue<>(N, new Comparator<Integer>() {
        @Override
        public int compare(Integer o1, Integer o2) {
            return o1.compareTo(o2);
        }
    });

    @Override
    protected void reduce(IntWritable key, Iterable<NullWritable> values, Reducer<IntWritable, NullWritable, NullWritable, IntWritable>.Context context) throws IOException, InterruptedException {
        topNQueue.add(key.get());
        if (topNQueue.size() > N) {
            topNQueue.poll();
        }
    }

    @Override
    protected void cleanup(Reducer<IntWritable, NullWritable, NullWritable, IntWritable>.Context context) throws IOException, InterruptedException {
        while (!topNQueue.isEmpty()) {
            context.write(NullWritable.get(), new IntWritable(topNQueue.poll()));
        }
    }
}
